package com.yanggu.redisson.component.producer;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.redisson.codec.TypedJsonJacksonCodec;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Redisson延迟队列生产者组件
 *
 * @param <T> 队列中的数据类型
 */
@Slf4j
public class DelayedQueueProducerComponent<T> {

    private final String name;

    private final ObjectMapper objectMapper;

    private final RDelayedQueue<T> delayedQueue;

    /**
     * 构造方法
     *
     * @param redissonClient   Redisson客户端
     * @param delayedQueueName 延迟队列名称
     * @param codec            序列化策略
     */
    public DelayedQueueProducerComponent(RedissonClient redissonClient,
                                         String delayedQueueName,
                                         TypedJsonJacksonCodec codec) {
        this.name = delayedQueueName;
        RBlockingQueue<T> blockingQueue = redissonClient.getBlockingQueue(delayedQueueName, codec);
        this.objectMapper = codec.getObjectMapper();
        this.delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
    }

    /**
     * 添加延迟消息
     *
     * @param data     消息
     * @param delay    延迟时间
     * @param timeUnit 时间单位
     */
    @SneakyThrows
    public void add(T data, long delay, TimeUnit timeUnit) {
        log.info("Add sync delayed message, queue: {}, message: {}, delay time: {}, timeUnit: {}", name, objectMapper.writeValueAsString(data), delay, timeUnit);
        delayedQueue.offer(data, delay, timeUnit);
    }

    /**
     * 异步添加延迟消息
     *
     * @param data     消息
     * @param delay    延迟时间
     * @param timeUnit 时间单位
     */
    @SneakyThrows
    public RFuture<Void> addAsync(T data, long delay, TimeUnit timeUnit) {
        log.info("Add async delayed message, queue: {}, message: {}, delay time: {}, timeUnit: {}", name, objectMapper.writeValueAsString(data), delay, timeUnit);
        return delayedQueue.offerAsync(data, delay, timeUnit);
    }

    /**
     * 删除延迟消息。延迟队列和阻塞队列都删除消息
     *
     * @param data 消息
     */
    @SneakyThrows
    public boolean remove(T data) {
        boolean remove = delayedQueue.remove(data);
        log.info("Remove delayed message, queue: {}, message: {}, result: {}", name, objectMapper.writeValueAsString(data), remove);
        return remove;
    }

    /**
     * 获取所有延迟消息
     *
     * @return 延迟消息列表
     */
    public List<T> readAll() {
        return delayedQueue.readAll();
    }

    /**
     * 清空延迟消息
     */
    public void clear() {
        delayedQueue.clear();
    }

}